
                /*********************************************
                **                                          **
                **                                          **
                **              g.d.nb.stop.C               **
                **                                          **
                **                                          **
                *********************************************/
                /*********************************************
                **                                          **
                **         Author Jean-Marc Adamo           **
                **       Copyright (C) 1997-2008 JMA        **
                *********************************************/


/*
Direct Access Global 
Non-Blocking Global Read/Write Functions
Polling is performed by local scheduler
Needs 3 processes
*/

#include "arch.h"

Global_id gid;
Coll_id cid1, cid2, cid3;

void init_gid(int, char **){
   gid.set();
   cid1.set();
   cid2.set();
   cid3.set();
   fprintf(Sdf::trace, "from node %d: gid.tag=%d,  cid1.tag=%d, cid2.tag=%d,  cid3.tag=%d\n",
           Sdf::PROC_ID, gid.get(), cid1.get(), cid2.get(), cid3.get());
   fflush(Sdf::trace);
}

void wcs_handler(HRW_rcd *);
void rcs_handler(HRW_rcd *);

int count;

class Root_thread: public Thread{
   public:
        Root_thread()
                : Thread((Tmf0_ptr)&Root_thread::body){}

        void body(){
           fprintf(Sdf::trace, "from node%d: this is my rank\n", Sdf::PROC_ID);
           fflush(Sdf::trace);
           char source_data[21];
           char rw_target[21];
           char dest_data[21];

           int s;
           for(s=0; s<21; s++)
              source_data[s]  = Sdf::PROC_ID;
           for(s=0; s<21; s++)
              rw_target[s]    = 50+Sdf::PROC_ID;
           for(s=0; s<21; s++)
              dest_data[s]    = 100+Sdf::PROC_ID;

           int i;
           for(i=0; i<21; i++){
             fprintf(Sdf::trace, "node%d: s[%d]=%d, t[%d]=%d, d[%d]=%d\n",
                     Sdf::PROC_ID, i, source_data[i], i, rw_target[i], i, dest_data[i]);
             fflush(Sdf::trace);
           }
           Global<char> global(gid, rw_target);
              //this is to ensure global completion of chan construction
           Sdf::barrier(cid1);

           count = 1;
           global.write((Sdf::PROC_ID+1)%Sdf::PROC_NB, 0, 
                        source_data[0], 21, wcs_handler, 0, 0);
           stop();

           Sdf::barrier(cid2);

           count = 1;
           global.read((Sdf::PROC_ID+Sdf::PROC_NB-1)%Sdf::PROC_NB, 0, 
                       dest_data[0], 21, rcs_handler, 0, 0);
           stop();

           for(i=0; i<21; i++)
             fprintf(Sdf::trace, "node%d: s[%d]=%d, t[%d]=%d, d[%d]=%d\n",
                     Sdf::PROC_ID, i, source_data[i], i, rw_target[i],i, dest_data[i]);
           Sdf::barrier(cid3);
        }
};

MAIN(Root_thread, init_gid, Sdf::empty, Sdf::empty)

void wcs_handler(HRW_rcd *ptr){
   if(!(count-=1))
      Sdf::ROOT_THREAD->schedule();

   fprintf(Sdf::trace, "node%d: global-writing's over on call side\n", Sdf::PROC_ID);
   fprintf(Sdf::trace, "node%d: wcs_handler, rem_proc=%d, d_ptr=%d, g-sz=%d, th_ptr=%d, hndlr_ptr=%d, hndlr_d_ptr=%d\n", 
           Sdf::PROC_ID,
           ptr->get_remote_proc(), ptr->get_data_ptr(),
           ptr->get_given_size(), ptr->get_thread_ptr(),
           ptr->get_hndlr_ptr(), ptr->get_hndlr_data_ptr());
   fflush(Sdf::trace);
}

void rcs_handler(HRW_rcd *ptr){
   if(!(count-=1))
      Sdf::ROOT_THREAD->schedule();

   fprintf(Sdf::trace, "node%d: global-reading's over on call side\n", Sdf::PROC_ID);
   fprintf(Sdf::trace, "node%d: rcs_handler, rem_proc=%d, d_ptr=%d, g-sz=%d, r_sz=%d, th_ptr=%d, hndlr_ptr=%d, hndlr_d_ptr=%d\n",
           Sdf::PROC_ID,
           ptr->get_remote_proc(), ptr->get_data_ptr(),
           ptr->get_given_size(), ptr->get_rcvd_size(),
           ptr->get_thread_ptr(), ptr->get_hndlr_ptr(),
           ptr->get_hndlr_data_ptr());
   fflush(Sdf::trace);
}
